package com.bigdata.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext, StreamingContextState}

/**
 * @author Gerry chan
 * @version 1.0
 *          2020/12/29 20:28
 * Spark Streaming关闭：
 * https://www.bilibili.com/video/BV11A411L7CK?p=199
 *
 * */
object SparkStreaming08_Close {
  def main(args: Array[String]): Unit = {
    /**
     * 线程的关闭
     * val thread = new Thread()
     * thread.start()
     *
     * thread.stop()  //强制关闭
     */
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
    val ssc = new StreamingContext(sparkConf, Seconds(3))

    val lines = ssc.socketTextStream("localhost", 9999)
    val wordToOne = lines.map((_,1))

    wordToOne.print()

    ssc.start()

    //如果想要关闭采集器，那么需要创建新的线程
    //而且需要在第三方程序中增加 关闭状态
    new Thread(
      new Runnable {
        override def run(): Unit = {
          //优雅的关闭，用于程序升级, 当前数据处理完成后再关闭
          while (true) {
            // 这里的true代表 从第三方获取的状态
            if (true) {
              //获取SparkStreaming状态
              val state: StreamingContextState = ssc.getState()
              if (state == StreamingContextState.ACTIVE) {
                ssc.stop(true, true)
                //停掉后 退出当前线程
                System.exit(0)
              }
            }
            Thread.sleep(50000)
          }

        }
      }
    ).start()

    ssc.awaitTermination() // block阻塞main线程


  }

}
